// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: wsfuyibing <682805@qq.com>
// Date: 2024-11-19

package logger

import (
	"context"
	"gitee.com/go-libs/log/base"
	"gitee.com/go-libs/log/common"
	"gitee.com/go-libs/log/conf"
	"gitee.com/go-libs/log/logger/elastic_plugin"
	"gitee.com/go-libs/log/logger/file_plugin"
	"gitee.com/go-libs/log/logger/kafka_plugin"
	"gitee.com/go-libs/log/logger/term_plugin"
	"gitee.com/go-libs/runtime"
	"sync"
	"sync/atomic"
	"time"
)

var (
	once      = new(sync.Once)
	singleton common.LoggingManager
)

type manager struct {
	bucket     common.Bucket[common.Item]
	mu         *sync.RWMutex
	plugin     common.LoggingPlugin
	processing int32
	sync       bool
}

// New
// 管理器单例.
func New() common.LoggingManager { return singleton }

// GetPlugin
// 读取插件实例.
func (o *manager) GetPlugin() (plugin common.LoggingPlugin) {
	o.mu.RLock()
	plugin = o.plugin
	o.mu.RUnlock()
	return
}

// Log
// 发送日志.
func (o *manager) Log(ctx context.Context, level base.Level, message string) {
	o.Send(common.NewItem(ctx, level, message))
}

// Send
// 发送日志.
func (o *manager) Send(items ...common.Item) {
	if o.sync {
		atomic.AddInt32(&o.processing, 1)
		defer atomic.AddInt32(&o.processing, -1)
		o.do(items...)
	} else {
		o.bucket.Add(items...)
	}
}

// SetPlugin
// 设置插件实例.
func (o *manager) SetPlugin(plugin common.LoggingPlugin) {
	if plugin != nil {
		o.mu.Lock()
		o.plugin = plugin
		o.sync = plugin.Sync()
		o.mu.Unlock()
	}
}

// Wait
// 等待完成.
func (o *manager) Wait() {
	if o.sync {
		for {
			if atomic.LoadInt32(&o.processing) < 1 {
				break
			}
			time.Sleep(time.Millisecond * 10)
		}
	}
	o.bucket.Stop().Wait()
}

// +------------------------------------------------------------------------------+
// | Access methods                                                               |
// +------------------------------------------------------------------------------+

func (o *manager) do(items ...common.Item) {
	var (
		count = len(items)
		err   error
	)

	// 1. 监听结束.
	defer func() {
		// 捕获异常.
		if r := recover(); r != nil {
			common.Debugger.Fatal("logger.manager.do panic: %v", r)
		}

		// 输出错误.
		if err != nil {
			common.Debugger.Fatal("logger.manager.do error: %v", err)
		}

		// 释放回池.
		for _, item := range items {
			item.Release()
		}
	}()

	// 2. 调用插件.
	if err = o.plugin.Write(items); err != nil {
		runtime.GetCounter().LoggingCounter().IncrFailedCount(int64(count))
	} else {
		runtime.GetCounter().LoggingCounter().IncrSucceedCount(int64(count))
	}
}

func (o *manager) init() *manager {
	o.mu = &sync.RWMutex{}
	return o.initPlugin().initBucket()
}

func (o *manager) initBucket() *manager {
	o.bucket = common.NewBucket[common.Item]("logger", o.do)
	o.bucket.Start()
	return o
}

func (o *manager) initPlugin() *manager {
	// 1. 插件适配.
	switch conf.New().Adapter {
	case base.AdapterFile:
		o.plugin = file_plugin.New()
	case base.AdapterElastic:
		o.plugin = elastic_plugin.New()
	case base.AdapterKafka:
		o.plugin = kafka_plugin.New()
	default:
		o.plugin = term_plugin.New()
	}

	// 2. 输出模式.
	o.sync = o.plugin.Sync()
	return o
}

func init() {
	once.Do(func() {
		singleton = (&manager{}).init()
	})
}
